﻿using System;
using System.Collections;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;

namespace ECMessageQueue
{
    public class MQServer : MessageQueue
    {
        private string ip = "0.0.0.0";
        private int port = 61050;
        private int maxconn = int.MaxValue;
        private Socket socket;
        private int buffersizekb = 1024 * 2;
        private Thread listenThread;

        public struct SERVER_INFO
        {
            public string bindIp;
            public int listenPort;
            public int maxConnections;
            public int bufferSize;
            public int currentConnections;
        }

        private struct CLIENT
        {
            public int id;
            public string token;
            public Socket socket;
            public Thread thread;
        }

        private Hashtable clients;



        /**
         * string bind_ip_address 绑定的本地IP地址，默认值=0.0.0.0
         * int listen_port 本地监听端口，默认值=61050
         * int buffer_size 缓冲区大小，单位KB，默认值2MB
         * int max_connections 最大连接数，默认int.maxvalue
         * */
        public MQServer(ECLogger.Logger logger = null) : base(logger)
        {
            initSocketServer();
        }

        public MQServer(string bind_ip_address, ECLogger.Logger logger = null) : base(logger)
        {
            ip = bind_ip_address;
            initSocketServer();
        }

        public MQServer(int listen_port, int buffer_size = 2 * 1024, ECLogger.Logger logger = null) : base(logger)
        {
            port = listen_port;
            buffersizekb = buffer_size;
            initSocketServer();
        }

        public MQServer(int max_connections, ECLogger.Logger logger = null) : base(logger)
        {
            maxconn = max_connections;
            initSocketServer();
        }

        public MQServer(string bind_ip_address, int listen_port, int buffer_size_kb, int max_connections, ECLogger.Logger logger = null) : base(logger)
        {
            ip = bind_ip_address;
            port = listen_port;
            buffersizekb = buffer_size_kb;
            maxconn = max_connections;
            initSocketServer();
            runSender();
        }

        private void initSocketServer()
        {
            try
            {
                socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
                socket.Bind(new IPEndPoint(IPAddress.Parse(ip), port));
                socket.Listen(maxconn);
                clients = new Hashtable();
                listenThread = new Thread(new ThreadStart(listenClientConnect));
                listenThread.Start();
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }

        private void listenClientConnect()
        {
            try
            {
                while (isRunning)
                {
                    CLIENT c = new CLIENT();
                    c.socket = socket.Accept();
                    c.id = 0;
                    do
                    {
                        c.id = new Random().Next();
                    }
                    while (clients.ContainsKey(c.id));
                    c.token = genToken();
                    clients.Add(c.id, c);
                    string msg = string.Format("{0},{1}", c.id, c.token);
                    c.socket.Send(Encoding.Default.GetBytes(msg));
                    c.thread = new Thread(new ParameterizedThreadStart(reciveFromClient));
                    c.thread.Start(c);
                    log.Debug("Client connected: ID[{0}] TOKEN[{1}]", c.id, c.token);
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }

        private void reciveFromClient(object client)
        {
            CLIENT c = (CLIENT)client;
            while (isRunning)
            {
                try
                {
                    var buff = new byte[buffersizekb * 1024];
                    c.socket.Receive(buff);
                    MQMESSAGE mqm = bytesToMsg(buff);
                    mqm.remote = c.id;
                    if (mqm == null) continue;
                    if (mqm.token != c.token)
                    {
                        FESSC_MESSAGE.Msg rqmsg = new FESSC_MESSAGE.Msg(mqm.content);
                        FESSC_MESSAGE.Msg rsmsg = new FESSC_MESSAGE.Msg(false);
                        rsmsg.setHeader(rqmsg.getRqOperationCode(), 0, c.id, mqm.token);
                        rsmsg.setRsCode("CLIENT-TOKEN-INVALIED");
                        mqm.content = rsmsg.ToString();
                        mqm.length = rsmsg.ToString().Length;
                        sendMessage(mqm);
                        continue;
                    }
                    putRecvQueue(mqm);
                    log.Debug("Recive Message from Client[{0}] Len[{1}]:\r\n{2}", c.id, mqm.length, mqm.content);
                }
                catch (Exception)
                {
                    c.socket.Close();
                    clients.Remove(c.id);
                }
            }
        }

        protected override void sendBySocket(MQMESSAGE mqm)
        {
            try
            {
                CLIENT c = (CLIENT)clients[mqm.remote];
                c.socket.Send(msgToBytes(mqm));
                log.Debug("Send Message to Client[{0}] Len[{1}]:\r\n{2}", c.id, mqm.length, mqm.content);
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }

        public SERVER_INFO getServerInfo()
        {
            try
            {
                SERVER_INFO srv = new SERVER_INFO();
                srv.bindIp = ip;
                srv.listenPort = port;
                srv.bufferSize = buffersizekb;
                srv.maxConnections = maxconn;
                srv.currentConnections = clients.Count;
                return srv;
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }

        public void close()
        {
            isRunning = false;
            senderThread.Abort();
            listenThread.Abort();
            stopSender();
            foreach (DictionaryEntry de in clients)
            {
                CLIENT c = (CLIENT)de.Value;
                if (c.thread != null) c.thread.Abort();
                if (c.socket != null) c.socket.Close();
            }
            socket.Close();
        }

    }
}
